package com.wayhome.srpingbootmybatis.kafka;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

@Slf4j
@Service
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    public void sendMessage(String topic, String message) {
        try {
            ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send(topic, message);
            listenableFuture.addCallback(
                    o -> log.info("kafka topic: {} 发送成功,", topic),
                    throwable -> log.info("kafka topic: {} 发送异常，getMessage={}", topic, throwable.getMessage(), throwable)
            );
        } catch (Exception e) {
            log.info("kafka topic: {} 发送异常，getMessage={}", topic,e.getMessage(), e);
        }

    }

}
